package com.atgugu.flink.chapter11.tvf;

import com.atgugu.flink.bean.Person;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.Table;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;

import java.time.Duration;

import static org.apache.flink.table.api.Expressions.$;

/**
 * @Author lzc
 * @Date 2022/4/12 8:59
 */
public class Flink04_TVF_GroupingSets_1 {
    public static void main(String[] args) {
        Configuration conf = new Configuration();
        conf.setInteger("rest.port", 2000);
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment(conf);
        env.setParallelism(1);
    
        DataStream<Person> stream = env
            .fromElements(
                new Person("a1", "b1", "c1", "d1", 1000L, 10),
                new Person("a2", "b2", "c2", "d2", 2000L, 20),
                new Person("a3", "b3", "c3", "d3", 6000L, 30),
                new Person("a4", "b4", "c4", "d4", 8000L, 40),
                new Person("a5", "b5", "c5", "d5", 9000L, 50),
                new Person("a6", "b6", "c6", "d6", 11000L, 60),
                new Person("a7", "b7", "c7", "d7", 12000L, 80),
                new Person("a8", "b8", "c8", "d8", 13000L, 80)
            )
            .assignTimestampsAndWatermarks(
                WatermarkStrategy
                    .<Person>forBoundedOutOfOrderness(Duration.ofSeconds(3))
                    .withTimestampAssigner((p, ts) -> p.getTs())
            );
        
        StreamTableEnvironment tEnv = StreamTableEnvironment.create(env);
        
        Table table = tEnv.fromDataStream(stream, $("a"), $("b"), $("c"), $("d"), $("ts").rowtime(), $("score"));
        tEnv.createTemporaryView("p", table);
        
        /*tEnv.sqlQuery("select " +
                          " a, b, c, d, window_start, window_end " +
                          " from table( tumble( table p, descriptor(ts), interval '5' second ) ) " +
                          " group by window_start, window_end, grouping sets(" +
                          " (a, b, c, d), " +
                          " (a, b, c   ), " +
                          " (a, b      ), " +
                          " (a         ), " +
                          " (          ) " +
                          ")")
            .execute()
            .print();*/
    
        /*tEnv.sqlQuery("select " +
                          " a, b, window_start, window_end " +
                          " from table( tumble( table p, descriptor(ts), interval '5' second ) ) " +
                          " group by window_start, window_end, rollup(a,b)")
            .execute()
            .print();*/
    
        tEnv.sqlQuery("select " +
                          " a, b,c, window_start, window_end " +
                          " from table( tumble( table p, descriptor(ts), interval '5' second ) ) " +
                          " group by window_start, window_end, cube(a,b,c)")
            .execute() // abc ab ac bc a b c " "
            .print();
        
    }
}
/*
 
 
 
 */